<?php
/*
 * Copyright (c) 2008 Sergey Bondari, bondari@bondari.net
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * - Redistributions of source code must retain the above copyright
 *   notice, this list of conditions and the following disclaimer.
 * - Redistributions in binary form must reproduce the above copyright
 *   notice, this list of conditions and the following disclaimer in the
 *   documentation and/or other materials provided with the distribution.
 * - The name of the author may not be used to endorse or promote products
 *   derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

namespace Mosaic\System;



/**
 * Request broker implementation
 *
 * A separate node process, that delivers and queues requests and responses
 * inside of a node and between nodes. Uses network event machine-like multiplexed
 * communication. Internally, request broker uses mapping of streams to to integers.
 *
 * @package System
 * @subpackage Node
 */
class RequestBroker extends GenericRequestBroker {

	/**
	 * Array of sockets to other nodes array(NodeName => SocketID)
	 * @var array 
	 */
	private $nodeSockets;


	/**
	 * Nameserver resolved requests cache
	 * @var array
	 */
	private $nsCache;


	/**
	 * Request broker server loop
	 * @return void
	 */
	public function run() {
		global $sys;
		$this->nodeSockets = array();
	
		/*
		 * Open node socket.
		 */
		trigger_error('Opening new node socket on '.$sys->config->get('node.listen'));
		$this->extSocket = stream_socket_server($sys->config->get('node.listen'), $errno, $errstr);

		if (!$this->extSocket) {
			throw new IOException('Socket creation failed. ('.$errno.'): '.$errstr);
		} else {
			trigger_error('External socket created');
		}

		/*
		 * Register known node sockets
		 */
		$this->extSocketID = $this->registerSocket($this->extSocket);
		$this->wtSocketID = $this->registerSocket($sys->wtSocket);
		$this->obSocketID = $this->registerSocket($sys->obSocket);
		$this->nsMsgSocketID = $this->registerSocket($sys->nsMsgSocket);

		try {
			while (1) {
				/*
				 * Always receive requests on the external listening socket
				 * and internal IPC sockets
				 */
				$socketsToRead = array($this->extSocket, $sys->wtSocket, $sys->obSocket, $sys->nsMsgSocket);
				
				/*
				 * Add all active nodes connections
				 */
				foreach ($this->nodeSockets as $socketID) {
					$socketsToRead[] = $this->socket($socketID);
				}

				/*
				 * Add all sockets from queue to write event
				 */
				$socketsToWrite = array();
				$_sids = array();
				foreach  ($this->queue as $_ => $queue) {
					if (!isset($_sids[$_]) && $queue->count()) $socketsToWrite[] = $this->socket($_);
				}

				trigger_error('+Request broker is entering select(). '.count($socketsToRead).' sockets to read. '.count($socketsToWrite).' sockets to write.');


				/*
				 * Wait for socket event
				 */
				$stat = stream_select($socketsToRead, $socketsToWrite, $except = NULL, NULL);

				/* Unexpected behaviour */
				if ($stat === false) {
					trigger_error('stream_select() did not return any sockets.',E_USER_ERROR);
					continue;
				}

				trigger_error('Incoming socket event');

				/*
				 * Process incoming requests
				 */
				foreach ($socketsToRead as $socket) {
					$socketID = $this->socketId($socket);
					/*
					 * New external request/response
					 */
					if ($socket == $this->extSocket) {
						/* Open new socket for this request */
						if (!$_newSocket = stream_socket_accept($socket)) {
							throw new IOException('Could not socket_accept()');
						}

						$socket = $_newSocket;
					} else $_newSocket = false;

					/*
					 * New internal request/response 
					 */
					if (feof($socket)) {
						trigger_error('Remote party has closed connection.');
						if ($_newSocket) {
							fclose($_newSocket);
						} else {
							$this->unregisterSocket($socketID);
						}
						if ($socket == $sys->wtSocket) {
							trigger_error('Worker has closed socket. Terminating.');
							return;
						}
						if ($socket == $sys->obSocket) {
							trigger_error('Object broker has closed socket. Terminating.');
							return;
						}
						if ($socket == $sys->nsMsgSocket) {
							trigger_error('Nameserver has closed control socket. Terminating.');
							return;
						}
						continue;

					}

					/* Parse request as MCGI */
					try {
						$mcgi = new \Mosaic\MCGI\Packet();
						$mcgi->receive($socket);
					} catch (IOException $e) {
						/* Log error. Close socket. */
						trigger_error('Error parsing MCGI request: '.$e->getMessage(), E_USER_ERROR);
						$this->unregisterSocket($socketID);
						continue;
					}

					if ($_newSocket) {
						$socketID = $this->registerSocket($_newSocket, $mcgi->protocol());
					}

					switch ($mcgi->protocol()) {
						case \Mosaic\MCGI\SCGI:
							$request = new HTTPContentRequest($mcgi);
							$this->processRequest($request, $socketID);
						break;

						case \Mosaic\MCGI\MCGI;
							if ($mcgi->getHeader('RESPONSE')) {
								$response = Response::createFromMCGIPacket($mcgi);
								$this->processResponse($response, $socketID);
							} else {
								$request = Request::createFromMCGIPacket($mcgi);
								$this->processRequest($request, $socketID);
							}
						break;
					}
				}

				/*
				 * Process outgoing requests
				 */
				foreach ($socketsToWrite as $socket) {
					$socketID = $this->socketId($socket);
					trigger_error('Socket '.$socketID.' is ready to receive data. Extracting request.');

					/*
					 * Queue is not empty. Otherwise we would not listen for write
					 */
					$r = $this->queue[$socketID]->extract();
					trigger_error('Preparing MCGI packet');
					$mcgi = $r->createMCGIPacket($this->socketProtocol($socketID));
					trigger_error('Sending MCGI packet');

					try {
						$mcgi->send($socket);
						/*
						 * SCGI protocol indicates that the request was from web server. We need to close socket and unregister.
						 */
						if ($mcgi->protocol() == \Mosaic\MCGI\SCGI) $this->unregisterSocket($socketID);
					} catch (IOException $e) {
						trigger_error('Socket did not accept all data. Aborting and unregistering.', E_USER_ERROR);
						$this->unregisterSocket($socketID);
					} catch (RuntimeException $e) {
						trigger_error('Runtime exception during send. Forcing silent socket unregistration.', E_USER_ERROR);
						try {
							$this->unregisterSocket($socketID);
						} catch (Exception $e) {
							trigger_error('Registration failed. Socket is already unregistered.', E_USER_WARNING);
						}

					}
				}
				
				// process

				trigger_error('-Socket event processed.');

			}


		} catch (Exception $e) {
			echo '[Exception] '.get_class($e).': '.$e->getMessage()."\n";
		}

		$this->unregisterSocket($this->extSocketID);

	}


	/**
	 * Unregister socket
	 *
	 * @param socket Socket to unregister
	 * @return void
	 */
	protected function unregisterSocket($socket) {
		if (is_resource($socket)) {
			$socket = $this->socketId($socket);
		}

		unset($this->nodeSockets[array_search($socket, $this->nodeSockets)]);
		parent::unregisterSocket($socket);
		
	}



	/**
	 * Incoming request processing routine
	 *
	 * Based on the request type function delivers request to the correct recipient that can be
	 * worker, object broker or other node's request broker.
	 *
	 * @param Request
	 * @param integer Socket which received the request
	 * @return void
	 */
	protected function processRequest(Request $req, $socketID) {
		global $sys;
		trigger_error('Processing request of type '. $req->getRequestType());

		$this->pool($socketID, $req);

		try {
			switch ($req->getRequestType()) {

				/*
				 * Request from web server
				 */
				case RequestType::HTTP_CONTENT:
					trigger_error('New HTTP content request');
					/*
					 * Route HTTP content request to the worker.
					 * Priority is lower than internal requests.
					 */
					$this->enqueue($this->wtSocketID, $req, 5);
				break;

				/*
				 * Request from control agent, announcing new node in site. Highest priority delivery
				 */
				case RequestType::NODE_UP_MANIFEST:
					trigger_error('Node up manifest received from "'.$req->getNodeName().'"');
					trigger_error('Forwarding to worker');
					$_req = new NodeUpManifest($req->getNodeName(),$req->getDSN(), $req->getManifest());
					$this->enqueue($this->wtSocketID, $_req, 10);
					trigger_error('Forwarding to object broker');
					$_req = new NodeUpManifest($req->getNodeName(),$req->getDSN(), $req->getManifest());
					$this->enqueue($this->obSocketID, $_req, 10);
				break;

				/*
				 * Request from control agent, announcing node left site. Highest priority
				 */
				case RequestType::NODE_DOWN_MANIFEST:
					trigger_error('Node down manifest received from "'.$req->getNodeName().'"');
					trigger_error('Forwarding to worker');
					$_req = new NodeDownManifest($req->getNodeName());
					$this->enqueue($this->wtSocketID, $_req, 10);
					trigger_error('Forwarding to object broker');
					$_req = new NodeDownManifest($req->getNodeName());
					$this->enqueue($this->obSocketID, $_req, 10);

					/* Clear NS cache */
					if (count($this->nsCache)) {
						foreach (array_search($req->getNodeName(), $this->nsCache) as $className) unset($this->nsCache[$className]);
					}
				break;

				/*
				 * Handshake from other node
				 */
				case RequestType::PING:
					$this->registerNode($req->getNodeName(), $socketID);
					$res = new Response($req);
					$res->setData($sys->name);
					$this->processResponse($res, $socketID);
				break;

				/*
				 * Request for resource. 
				 * Depending on the destination will be delivered either to my worker 
				 * or to other node's request broker.
				 */
				case RequestType::RESOURCE:
					$node = $req->getDestination();
					if ($node == $sys->name) {
						trigger_error('New resource request for my worker!');
						$this->enqueue($this->wtSocketID, $req, 5);
					} else {
						trigger_error('Request is for node name '.$node);
						if (isset($this->nodeSockets[$node])) {
							trigger_error('Node '.$node.' is known to me.');
						} else {
							$this->registerNode($node);
						}

						$this->enqueue($this->nodeSockets[$node], $req, 5);

					}
				break;

				/*
				 * Remote object instantiation request
				 *
				 * Depending on the class name will be delivered either to my object broker or other node's request broker. 
				 * Nameservice is asked for the node name responsible for objects of this type. NS results are cached.
				 */
				case RequestType::REMOTE_OBJECT_INSTANTIATE:

					if (RB_NO_CACHE||!isset($this->nsCache[$req->getClassName()])) {
						$_r = new RemoteClassResolveRequest($req->getClassName());
						$res = $_r->sendReceive($sys->nsSocket);
						$node = $res->getData();
						$this->nsCache[$req->getClassName()] = $node;
					} else {
						$node = $this->nsCache[$req->getClassName()];
					}

					if ($node == $sys->name) {
						trigger_error('New request for my object broker.');
						$this->enqueue($this->obSocketID, $req, 9);
					} else {
						trigger_error('Request is for node name '.$node);
						if (isset($this->nodeSockets[$node])) {
							trigger_error('Node '.$node.' is known to me.');
						} else {
							$this->registerNode($node);
						}

						$this->enqueue($this->nodeSockets[$node], $req, 9);

					}
				break;

				/*
				 * Remote object invocation request
				 *
				 * Depending on the class name will be delivered either to my object broker or other node's request broker. 
				 * Nameservice is asked for the node name responsible for objects of this type. NS results are cached.
				 */
				case RequestType::REMOTE_OBJECT_CALL:
					if (!preg_match('#^mcgi://([^/]+)/([^/]+)/(.+)$#', $req->getUri(), $regs)) {
						throw new RuntimeException('Invalid URI format for remote object ['.$req->getUri().']');
					}

					$nodeName = $regs[1];
					$className = $regs[2];


					if (RB_NO_CACHE||!isset($this->nsCache[$className])) {
						$_r = new RemoteClassResolveRequest($className);
						$res = $_r->sendReceive($sys->nsSocket);
						$node = $res->getData();
						$this->nsCache[$className] = $node;
					} else {
						$node = $this->nsCache[$className];
					}

					if ($nodeName != $node) {
						throw new RuntimeException('Invalid URI. Outdated node is no longer accepts this class ['.$req->getUri().']');
					}

					if ($node == $sys->name) {
						trigger_error('New request for my object broker!');
						$this->enqueue($this->obSocketID, $req, 5);
					} else {
						trigger_error('Request is for node name '.$node);
						if (isset($this->nodeSockets[$node])) {
							trigger_error('Node '.$node.' is known to me.');
						} else {
							$this->registerNode($node);
						}
						$this->enqueue($this->nodeSockets[$node], $req, 5);
					}

				break;

				/*
				 * Remote object stub code request
				 *
				 * Depending on the class name will be delivered either to my object broker or other node's request broker. 
				 * Nameservice is asked for the node name responsible for objects of this type. NS results are cached.
				 */
				case RequestType::REMOTE_OBJECT_STUB:
					$className = $req->getClassName();

					/*
					 * Get the node name for the class name
					 */
					if (RB_NO_CACHE||!isset($this->nsCache[$className])) {
						$_r = new RemoteClassResolveRequest($className);
						$res = $_r->sendReceive($sys->nsSocket);
						$node = $res->getData();
						$this->nsCache[$className] = $node;
					} else {
						$node = $this->nsCache[$className];
					}

					/*
					 * If mine, deliver to object broker
					 */
					if ($node == $sys->name) {
						trigger_error('New request for my object broker!');
						$this->enqueue($this->obSocketID, $req, 100);
					} else {
						trigger_error('Request is for node name '.$node);
						if (isset($this->nodeSockets[$node])) {
							trigger_error('Node '.$node.' is known to me.');
						} else {
							$this->registerNode($node);
						}
	
						$this->enqueue($this->nodeSockets[$node], $req, 100);
					}

				break;


				/*
				 * Release context. Broadcasted to all other known nodes to me
				 */
				case RequestType::RELEASE_CONTEXT:
					unset($this->pool[$req->getRequestToken()]);
					if ($socketID == $this->wtSocketID) { // from my worker => deliver to other nodes
						foreach ($this->nodeSockets as $node => $socket) {
							$this->enqueue($socket, $req, 1);
						}
					} else { // from external socket - delivery to my worker
						$this->enqueue($this->wtSocketID, $req, 1);
					}
				break;


			default:
				throw new RuntimeException('Invalid request type');

			}
		} catch (RemoteExceptionThrowable $e) {

			/**
			 * Deliver app error from request broker to sender
			 */
			$response = new Response($req);
			$response->setData($e);
			$this->processResponse($response, $socketID);
		}

	}



	/**
	 * Other node registration
	 *
	 * Registers node with this broker. This function is called from different places of broker. If remote node initated connection
	 * socket already exists and we just add socket to the pool. If we are initiating the connection, we need to get node DSN from NS
	 * facility, open connection to the node and perform handshake.
	 *
	 * @param string Name of the node
	 * @param integer Socket of connection to the node. If parameter is ommited, function will establish connection.
	 * @return void
	 */
	private function registerNode($nodeName, $socketID = false) {
		global $sys;
		trigger_error('+Registering node '.$nodeName);

		if ($socketID === false) {

			$_r = new NodeResolveRequest($nodeName);
			trigger_error('Resolving node name.');
			$res = $_r->sendReceive($sys->nsSocket);
			$socketAddress = $res->getData();

			trigger_error('Connecting to node.');
			$socket = stream_socket_client($socketAddress,$errno,$errstr);

			if (!$socket) {
				throw new RuntimeException('Could not register node '.$nodeName.' on '.$socketAddress);
			}

			trigger_error('Sending handshake.');
			$req = new PingRequest($sys->name);
			$response = $req->sendReceive($socket);
			if ($response->getData() == $nodeName) {
				trigger_error('Handshake success');
			} else {
				throw new RuntimeException('Handshake failed with '.$nodeName);
			}
		
			$this->nodeSockets[$nodeName] = $this->registerSocket($socket);
		} else {
			$this->nodeSockets[$nodeName] = $socketID;
		}

		trigger_error('-Node registered.');

	}

	/**
	 * Class destructor
	 */
	function __destroy() {
		if ($this->extSocket) fclose($this->extSocket);
	}

}
